package dacp.etl.kafka.hdfs.connect.names;

import java.util.Date;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.kafka.common.TopicPartition;
import org.joda.time.DateTime;

import com.google.common.base.Splitter;
import com.google.common.collect.Maps;

import dacp.etl.kafka.hdfs.connect.model.FileNamePropertiesVO;
import dacp.etl.kafka.hdfs.connect.utils.CommonSinkConnectorConfig;

public class SCNames implements FileName{
	//兼容旧系统
	private final String PARTITION = "partition";
	private final String ONLYID = "onlyid";
	private final String RANDOM = "random";
	private final String BEGINMARK = "{";
	private final String ENDMARK= "}";
	private final String DIREVALSPLIT= "=";
	
	class DirAndCounter{
		private String directory; 
		private AtomicLong counter; 
		
		public DirAndCounter(String directory, AtomicLong counter){
			this.counter = counter;
			this.directory = directory;
		}
	}
	
 	private Map<Object, DirAndCounter> maps2 = Maps.newConcurrentMap();
	 
    private String nameFormat;  
    
    private Splitter spliter = Splitter.on('/'); 

	@Override
	public void configure(Map<String, Object> config) {
		nameFormat = (String) config.get(CommonSinkConnectorConfig.HDFS_FILENAME_FORMAT) ;
	}
	
	private long getRandom(String onlyId, String directory){
		DirAndCounter dc = maps2.get(onlyId);
		
		if(dc == null){
			String begin = new DateTime(new Date()).toString("HHmmss");
			long bl = Long.parseLong(begin) * 100;
			maps2.put(onlyId, new DirAndCounter(directory, new AtomicLong(bl))); //
		}else{
			if (!dc.directory.equals(directory)) {
				dc.counter = new AtomicLong(0);
				dc.directory = directory;
			}
		}
 		return maps2.get(onlyId).counter.incrementAndGet();
	} 
	
	@Deprecated
	private long getRandom(int partition, String directory){
		DirAndCounter dc = maps2.get(partition);
		
		if(dc == null){
			String begin = new DateTime(new Date()).toString("HHmmss");
			long bl = Long.parseLong(begin) * 100;
			maps2.put(partition, new DirAndCounter(directory, new AtomicLong(bl))); //
		}else{
			if (!dc.directory.equals(directory)) {
				dc.counter = new AtomicLong(0);
				dc.directory = directory;
			}
		}
 		return maps2.get(partition).counter.incrementAndGet();
	} 
	 
	
	public String fmtDirctory(String directory, final String fmt){
		String rst =  fmt;
		for(String s: spliter.split(directory)){
			if(s.contains(DIREVALSPLIT)){
				int indexOf = s.indexOf(DIREVALSPLIT); 
				rst = rst.replace(BEGINMARK + s.substring(0, indexOf) + ENDMARK, s.substring(indexOf+1));
			}
		} 
		return rst;
	}
	
	private String fmtReplace(String comm, Object o, final String fmt){
		String rst =  fmt;
		if(rst.contains(comm)){
			int b = rst.indexOf("{"+comm);
			int e = rst.indexOf("}", b);
			String a = fmt.substring(b+1, e);
			String f = a.substring(a.indexOf("%"));
			String format = String.format(f, o);
			rst = rst.substring(0, b) + format + rst.substring(e+1);
		}
		return rst;
	}
	 
	@Deprecated
	private String fmtPartition(int partition, final String fmt){
		return fmtReplace(PARTITION, partition,  fmt); 
	}
	
	@Deprecated
	private boolean isValidPartionId(String onlyId ){
		boolean rst = false;
		try{
			Integer.valueOf(onlyId);
			rst = true;
		}catch(Exception e){
			//this is ok
		}
		return rst;
	}
	
	public String fmtOnlyId(String onlyId, final String rst){
		String fmtRst = rst;
		if(isValidPartionId(onlyId)){
			fmtRst = fmtPartition(Integer.valueOf(onlyId), fmtRst);
		} 
		if(fmtRst.contains(PARTITION)){
			int inx = fmtRst.indexOf(BEGINMARK+PARTITION);
			int endInx = fmtRst.indexOf(ENDMARK, inx);
			String substring = fmtRst.substring(inx, endInx+1);
			fmtRst = fmtRst.replace(substring, onlyId);
		}
		return fmtRst.replace(BEGINMARK + ONLYID + ENDMARK, onlyId);
	}
	
	
	private String fmtRandom(long random, final String fmt){
		return fmtReplace(RANDOM, random,  fmt);   
	}
	
	public void setNameFmt(String fmt){
		this.nameFormat = fmt;
	}

	@Override
	public String generateName(TopicPartition topicPart, long startOffset, long endOffset, String directory, String extension) {
		int partition = topicPart.partition(); 

		String rst = nameFormat;
		rst = fmtDirctory(directory, rst);
		rst = fmtPartition(partition, rst);
		rst = fmtRandom(getRandom(partition, directory), rst);
		
		return rst;
	} 

	@Override
	public String generateName(FileNamePropertiesVO vo) {
		String fmt = nameFormat;
		String rst = fmtDirctory(vo.getDirectory(), fmt);
		rst = fmtOnlyId(vo.getOnlyId(), rst);
		rst = fmtRandom(getRandom(vo.getOnlyId(), vo.getDirectory()), rst);
		return rst;
	}

	public static void main(String[] args) {
		
		
		SCNames scNames = new SCNames();
		System.out.println(scNames.getRandom(100, "/1/2"));
		scNames.setNameFmt("H110001{hour_id}{partition%02d}{random%09d}");
		TopicPartition topicPart = new TopicPartition("LETMM1", 992);
		
		String generateName = scNames.generateName(topicPart, 0L, 0L, "/a/b/hour_id=201102012x", ".AVL");
		System.out.println(generateName);
	}


	
}
